package com.atguigu.source;

import com.atguigu.been.WaterSensor;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.Random;

/**
 * @author wky
 * @create 2021-07-13-16:37
 */

//自定义source
public class Source_Custom {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment senv = StreamExecutionEnvironment.getExecutionEnvironment();
        //TODO 2.通过自定义数据源获取数据
        DataStreamSource<WaterSensor> streamSource = senv.addSource(new MySource());
        streamSource.print();
        senv.execute();

    }

    //TODO 自定义source
    public static class MySource implements SourceFunction<WaterSensor>{

        private Random random = new Random();
        boolean bool = true ;

        @Override
        public void run(SourceContext<WaterSensor> sourceContext) throws Exception {
            while (bool){
                sourceContext.collect(new WaterSensor("水位"+random.nextInt(9)+1,System.currentTimeMillis(),random.nextInt(100)+100));
                Thread.sleep(1000);
            }
        }

        @Override
        public void cancel() {
            bool=false;
        }
    }
}

